package com.pinfly.common.http;

import java.io.IOException;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.log4j.Logger;

/**
 * A pool responsible for closing Sockets. The pool uses multiple threads (up to
 * a configured max) to perform the closes. Threads are only started as needed.
 * If a Thread is idle for a specific time period, the thread is shutdown.
 * 
 */
class SocketClosePool
{
    private static final Logger logger = Logger.getLogger (SocketClosePool.class);

    // Threads used to close Sockets
    private List m_workerThreads = new ArrayList ();

    // A Queue of Sockets to close
    private List m_closeQ = new ArrayList ();

    private final Thread m_monitorThread = new SocketCloseThreadMon ();
    private int m_maxThreads;
    private int m_idleTimeout;

    /**
     * Creates a new instance
     * 
     * @param maxThreads The maximum number of concurrent threads to use for
     *            closing Sockets
     * @param idleTimeout How long (in msecs) an idle thread can exist in the
     *            pool before it is shutdown.
     * @throws IllegalArgumentException if maxThreads is less than 1
     * @throws IllegalArgumentException if idleTimeout is less than 0
     */
    SocketClosePool (int maxThreads, int idleTimeout)
    {
        if (maxThreads < 1)
        {
            throw new IllegalArgumentException ("maxThreads must be greater than zero!");
        }

        if (idleTimeout < 0)
        {
            throw new IllegalArgumentException ("idleTimeout must be greater than -1!");
        }

        m_maxThreads = maxThreads;
        m_idleTimeout = idleTimeout;

        // Start up a thread to monitor the pool and shutdown
        // idle threads.
        m_monitorThread.setDaemon (true);
        m_monitorThread.start ();
    }

    /**
     * Gets the size of the Worker pool, which is the number of running threads
     * available to process close requests.
     * 
     * @return the size of the pool
     */
    int getWorkerPoolSize ()
    {
        synchronized (m_workerThreads)
        {
            return m_workerThreads.size ();
        }
    }

    /**
     * Gets the size of the close queue, which contains Sockets that are waiting
     * to be closed.
     * 
     * @return the size of the pool
     */
    int getCloseQueueSize ()
    {
        synchronized (m_closeQ)
        {
            return m_closeQ.size ();
        }
    }

    /**
     * Shuts down all of the worker threads, and removes any pending requests in
     * the close Q.
     * 
     */
    void shutDownPool ()
    {
        m_monitorThread.interrupt ();

        try
        {
            m_monitorThread.join ();
        }
        catch (InterruptedException ignored)
        {
        }

        synchronized (m_workerThreads)
        {
            for (Iterator iter = m_workerThreads.iterator (); iter.hasNext ();)
            {
                SocketCloseThread w = (SocketCloseThread) iter.next ();
                w.shutDown ();
                iter.remove ();
            }
        }

        synchronized (m_closeQ)
        {
            m_closeQ.clear ();
        }
    }

    /**
     * Adds a Socket to be closed
     * 
     * @param socket The socket to be closed.
     */
    void add (Socket sock)
    {
        int qSize;

        // Put it in a Q
        synchronized (m_closeQ)
        {
            m_closeQ.add (sock);
            qSize = m_closeQ.size ();
            m_closeQ.notify ();
        }

        // Add a new worker thread if:
        // - there are no worker threads
        // - there are more than a few records in the queue and we
        // haven't reached our maximum number of worker threads.
        synchronized (m_workerThreads)
        {
            int size = m_workerThreads.size ();
            if (size == 0 || (qSize > 4 && size < m_maxThreads))
            {
                SocketCloseThread t = new SocketCloseThread ();
                t.setDaemon (true);
                t.start ();
                m_workerThreads.add (t);
            }
        }

    }

    /**
     * A thread responsible for closing Socket instances.
     */
    private class SocketCloseThread extends Thread
    {
        private volatile boolean m_run = true;
        private long m_lastCloseTime = System.currentTimeMillis ();

        private SocketCloseThread ()
        {
            super ("SocketCloseThread from " + Thread.currentThread ().getName ());
        }

        private void shutDown ()
        {
            m_run = false;
            interrupt ();
        }

        private boolean expired ()
        {
            return (System.currentTimeMillis () > (m_lastCloseTime + m_idleTimeout));
        }

        public void run ()
        {
            while (m_run)
            {
                Socket sock = null;

                synchronized (m_closeQ)
                {
                    if (m_closeQ.size () == 0)
                    {
                        try
                        {
                            m_closeQ.wait ();
                        }
                        catch (InterruptedException e)
                        {
                            // Someone is possibly trying to stop us, if that's
                            // true m_run will be false and we will exit.
                            continue;
                        }
                    }

                    if (m_closeQ.size () > 0)
                    {
                        sock = (Socket) m_closeQ.remove (m_closeQ.size () - 1);
                    }
                }

                if (sock != null)
                {
                    m_lastCloseTime = System.currentTimeMillis ();

                    try
                    {
                        sock.close ();
                    }
                    catch (IOException ignored)
                    {
                    }

                    if (logger.isTraceEnabled ())
                    {
                        logger.trace ("Took " + (System.currentTimeMillis () - m_lastCloseTime)
                                      + " msecs to close the following Socket:\n" + sock);
                    }
                }
            }
        }
    }

    // A thread responsible for monitoring idle worker threads
    // and closing them if they have been idle too long
    private class SocketCloseThreadMon extends Thread
    {
        private SocketCloseThreadMon ()
        {
            super ("SocketCloseThreadMon from " + Thread.currentThread ().getName ());
        }

        public void run ()
        {
            // Runs until interrupted
            while (!isInterrupted ())
            {
                try
                {
                    sleep (15000);
                }
                catch (InterruptedException ex)
                {
                    // someone trying to shut us down
                    break;
                }

                synchronized (m_workerThreads)
                {
                    for (Iterator iter = m_workerThreads.iterator (); iter.hasNext ();)
                    {
                        SocketCloseThread t = (SocketCloseThread) iter.next ();
                        if (t.expired ())
                        {
                            t.shutDown ();
                            iter.remove ();
                        }
                    }
                }
            }
        }
    }

}
